package com.atguigu.userprofile.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.userprofile.bean.TagCondition;
import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.bean.UserGroup;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.mapper.UserGroupMapper;
import com.atguigu.userprofile.service.TagInfoService;
import com.atguigu.userprofile.service.UserGroupService;
import com.atguigu.userprofile.utils.RedisUtil;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;

import org.apache.catalina.User;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2021-05-04
 */
@Service
@Slf4j
@DS("mysql")
public class UserGroupServiceImpl extends ServiceImpl<UserGroupMapper, UserGroup> implements UserGroupService {


    /*
            把页面传入的userGroup写入Mysql
                    主要为了让Mysql生成自增Id
     */
    @Override
    public void writeUserGroupToMysql(UserGroup userGroup) {

        //createtime
        Date date = new Date();

        userGroup.setCreateTime(date);

        //生成页面回显的条件注释
        String conditionJsonToComment = userGroup.conditionJsonToComment();
        userGroup.setConditionComment(conditionJsonToComment);

        //生成条件的json
        userGroup.setConditionJsonStr(JSON.toJSONString(userGroup.getTagConditions()));

        //写入Mysql
        save(userGroup);

    }

    @Override
    public void writeUserGroupToCK(UserGroup userGroup) {

        String bitmapAndSql = genetateBitmapAndSql(userGroup);

        super.baseMapper.insertUserGroupToCK(userGroup.getId().toString(),bitmapAndSql);

    }

    @Override
    public UserGroup writeUserGroupToRedis(UserGroup userGroup) {

        String id = userGroup.getId().toString();
        //查询ck中刚刚写入的人群包
        List<String> uids = super.baseMapper.queryUserGroupById(id);

        String [] value = new String[]{};

        value = uids.toArray(value);

        /*
            写入到redis
                K:  唯一标识    usergroup_{id}
                V:  多值类型
                        list:  允许重复
                        set :  不允许重复.
                                sismember: 可以判断某个人是否符合这个人群
                        zset :  排序的set
         */
        Jedis jedis = RedisUtil.getJedis();

        String key = "usergroup_" + id;

        //写新的数据之前，先删除旧数据。
        jedis.del(key);

        jedis.sadd(key,value);

        jedis.close();

        //为UserGroup的分群人数属性赋值
        userGroup.setUserGroupNum(uids.size()+0l);

        return userGroup;

    }

    @Override
    public Long evaluateUserGroupNum(UserGroup userGroup) {

        String bitmapAndSql = genetateBitmapAndSql(userGroup);

        Long num = super.baseMapper.evaluateUserGroupNum(bitmapAndSql);


        return num;
    }

    @Override
    public void updateCK(UserGroup userGroup) {

        String id = userGroup.getId().toString();
        //删除旧的
        super.baseMapper.deleteOldUserGroupById(id);

        //写入新的
        writeUserGroupToCK(userGroup);
    }

    @Autowired
    private TagInfoService tagInfoService;
    /*
        生成所有条件的查询:
                两两求交集
                 select bitmapAnd(
                        bitmapAnd(
                            (sql1) , (sql2)
                        ),
                        (sql3)
                   )

                数量关系:  bitmapAnd 的数量 =  sql的数量 - 1

                特殊情况： 只有一个条件。 只有一个sql,需要在外层添加()
                            select (select xxx from )
     */
    public String genetateBitmapAndSql(UserGroup userGroup){


        //获取所有的条件
        List<TagCondition> tagConditions = userGroup.getTagConditions();

        //所有三级标签的元数据
        Map<String, TagInfo> tags = tagInfoService.getTagInfoMapWithCode();

        //创建一个拼接的最终结果
        StringBuilder result = new StringBuilder();

        /*
              第一次遍历:  (sql1)
              第二次遍历:  bitmapAnd((sql1) ,(sql2))
              第三次遍历:  bitmapAnd(
                             bitmapAnd((sql1) ,(sql2))
                             ,(sql3)
                         )
         */
        for (TagCondition tagCondition : tagConditions) {

            String singleSql = generateTagConditionSql(tagCondition, tags, userGroup.getBusiDate());

            //判断是否是第一个条件，如果是需要在外层添加()
            if (result.length() == 0){
                result.append( "(" + singleSql +" )");
            }else {
                result.insert(0,"bitmapAnd(")
                      .append(",(")
                      .append(singleSql)
                      .append("))");
            }

        }

        return result.toString();

    }

    /*
        生成单个条件的子查询

        select
                groupBitmapMergeState(us)
         from user_tag_value_string     根据查询的标签的tagValueType确定从ck的哪张表查询。
                                        需要查询Mysql的元数据，得知标签的tagValueType。

                                        TagInfoServiceImpl.getTagInfoMapWithCode(): 查询出所有的三级标签的元数据
                                            Map<String,TagInfo>
                                                    key: tagCode
                                                    value: TagInfo


                                    举例:
                                        gender, string, user_tag_value_string
                                        7damount, decimal ,user_tag_value_decimal

         where tag_code = 'tag_population_attributes_nature_gender'
                                从 tagCondition中的 tagCode获取标签编码填入

           and tag_value
                        in         从 tagCondition中的 operator 中取。
                                    还需要转换  eq 转换 sql中的 =
                                    如果运算符号是in 或 notin，还需要再tagValues 外层拼接 ()

                        ('男性','女性')     从 tagCondition中的 tagValues中取。
                                        如果时多个值，需要进行拼接。
                                        如果查询的tagCode的tagValueType是String，还需要拼接''


           and dt = '2020-06-14'
                                从业务日期中取
     */
    public String generateTagConditionSql(TagCondition tagCondition,Map<String,TagInfo> tags,String doDate){

        String template =" select  groupBitmapMergeState(us)  from %s   where tag_code = '%s' " +
                         "  and tag_value %s %s   and dt = '%s' ";

        String tagCode = tagCondition.getTagCode();
        TagInfo tagInfo = tags.get(tagCode);

        //根据tagCode的tagValueType确定查询的ck的源表名
        String sourceTable = "";

        switch (tagInfo.getTagValueType()){
            case ConstCodes.TAG_VALUE_TYPE_LONG: sourceTable="user_tag_value_long"; break;
            case ConstCodes.TAG_VALUE_TYPE_DECIMAL: sourceTable="user_tag_value_decimal"; break;
            case ConstCodes.TAG_VALUE_TYPE_STRING: sourceTable="user_tag_value_string"; break;
            case ConstCodes.TAG_VALUE_TYPE_DATE: sourceTable="user_tag_value_date"; break;
        }

        //过滤时使用小写
        // 将前台传入的运算符 转换为 sql的运算符
        String sqlOperator = getConditionOperator(tagCondition.getOperator());

        //拼接条件
        String tagValueStr = "";

        List<String> tagValues = tagCondition.getTagValues();

        //如果tagValueType是String或Date要在参数之间拼接 引号
        if (ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagInfo.getTagValueType())
            ||
            ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagInfo.getTagValueType())){

            // 男性','女性','未知
            tagValueStr =  "'" + StringUtils.join(tagValues,"','") + "'";
        }else{
            // 2000,3000
            tagValueStr =   StringUtils.join(tagValues,',');
        }

        //如果运算符是 范围查询： in not in，还需要在条件最外层拼接 ()
        if ("in".equals(tagCondition.getOperator()) || "nin".equals(tagCondition.getOperator())){
            tagValueStr = "(" + tagValueStr + ")";
        }

        //格式化
        String sql = String.format(template, sourceTable, tagCode.toLowerCase(), sqlOperator, tagValueStr, doDate);

        return sql;

    }

    private String getConditionOperator(String operator){

        switch (operator) {
            case "eq" : return "=";
            case "neq" : return "<>";
            case "gte" : return ">=";
            case "lte" : return "<=";
            case "gt" : return ">";
            case "lt" : return "<";
            case "in" : return "in";
            case "nin" : return "not in";
        }
        throw new RuntimeException("没有匹配的操作符!");

    }
}
